package com.zf.nospring.kafka.simple;

import com.zf.nospring.kafka.utils.KafkaUtils;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 使用KafkaProducer发送消息
 */
public class ProducerTest {

    /**
     * 测试发送，同步获取发送的结果
     */
    @Test
    public void testSend() throws ExecutionException, InterruptedException {
        // 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = KafkaUtils.getDefaultProducer();
        // 调用 send 方法,发送消息
        for (int i = 1; i <= 200; i++) {
            // 头信息
            RecordHeaders headers = new RecordHeaders();
            headers.add(new RecordHeader("hk" + i, ("hv" + i).getBytes(StandardCharsets.UTF_8)));
            // key
            String key = "k" + i;
            // msg
            String msg = "测试消息 " + i;
            ProducerRecord<String, String> record = new ProducerRecord<>("test", null, key, msg, headers);
            Future<RecordMetadata> send = kafkaProducer.send(record);
            // 等待发送的结果
            RecordMetadata recordMetadata = send.get();
            System.out.println(recordMetadata);
        }
        // 关闭资源
        kafkaProducer.close();
    }

    /**
     * 测试发送，使用回调获取发送的结果
     */
    @Test
    public void testSendCallBack() {
        KafkaProducer<String, String> kafkaProducer = KafkaUtils.getDefaultProducer();
        // 回调
        Callback callback = new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                // 有异常，发送失败
                if (exception != null) {
                    System.out.println("发送失败");
                } else {
                    System.out.printf("%s 发送成功\n", metadata);
                }
            }
        };
        for (int i = 1; i <= 10; i++) {
            kafkaProducer.send(new ProducerRecord<>("test", "回调方式测试消息发送结果" + i), callback);
        }
        kafkaProducer.close();
    }

}
